package com.sf.fy.uav.applet.config;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;

import com.sf.kafka.api.produce.IKafkaProducer;
import com.sf.kafka.api.produce.KeyedString;
import com.sf.kafka.api.produce.ProduceConfig;
import com.sf.kafka.api.produce.ProduceOptionalConfig;
import com.sf.kafka.api.produce.ProducerPool;

/**
 * 消费基站上行数据;
 *
 * @author 854277
 */

@Component
@Scope("singleton")
public class GcmPorducer implements InitializingBean, DisposableBean {
    private static final Logger logger = LoggerFactory.getLogger(GcmPorducer.class);

    private IKafkaProducer kafkaProducer;

    @Value("${config.productPoolsize}")
    private int poolSize;

    @Value("${config.productSystemurl}")
    private String systemUrl;

    @Value("${config.productClustername}")
    private String clusterName;

    @Value("${config.report_gcm_topic}")
    private String topic;

    @Value("${config.product_gcm_TopicToken}")
    private String checkCode;


    @Override
    public void destroy() throws Exception {
        if (kafkaProducer != null) {
            kafkaProducer.close();
        }
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        final String topicTokens = topic + ":" + checkCode;
        ProduceConfig produceConfig = new ProduceConfig(poolSize, systemUrl, clusterName, topicTokens);
        ProduceOptionalConfig optionalConfig = new ProduceOptionalConfig();
        optionalConfig.setRequestRequiredAck(ProduceOptionalConfig.RequestRequiredAck.LEADER_REPLICA);
        kafkaProducer = new ProducerPool(produceConfig, optionalConfig);
    }

    /**
     * 发送消息
     * <p>
     * 指定分片发送消息
     */

    public void sendString(String message) {
        logger.info(String.format("kafka product topic %s, message %s", this.topic, message));
        kafkaProducer.sendString(this.topic, message);
    }

    /**
     * 指定分片发送消息
     */
    public void sendKeyedString(KeyedString message) {
        // logger.info(String.format("kafka product topic %s, key %s, message %s", this.topic, ((DisKeyedString) message).getKey(), ((DisKeyedString) message).getMessage()));
        kafkaProducer.sendKeyedString(this.topic, message);
    }


}
